package com.junmoo.www.config.kafka;//package com.junmo.emqx.kafka;

import com.alibaba.fastjson.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Component;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.ListenableFutureCallback;

import java.util.function.Consumer;

@Component
@Slf4j
public class KafkaProducer {
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    //默认主题
    public static final String TOPIC_DEFAULT = "topic.default";

    public void send(String topic , String key, Object payload, Consumer<SendResult<String, Object>> success, Consumer<Throwable> throwable) {
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic,key, payload);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable e) {
                throwable.accept(e);
            }

            @Override
            public void onSuccess(SendResult<String, Object> sendResult) {
                // 处理消息发送成功逻辑
                success.accept(sendResult);
            }
        });
    }

    public void send(String topic ,Object payload, Consumer<SendResult<String, Object>> success, Consumer<Throwable> throwable) {
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, payload);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable e) {
                throwable.accept(e);
            }

            @Override
            public void onSuccess(SendResult<String, Object> sendResult) {
                // 处理消息发送成功逻辑
                success.accept(sendResult);
            }
        });
    }

    public void send(Object payload, Consumer<SendResult<String, Object>> success, Consumer<Throwable> throwable) {
        String obj2String = JSONObject.toJSONString(payload);
        log.info("------准备发送消息为：{}", obj2String);
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_DEFAULT, payload);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable e) {
                throwable.accept(e);
            }

            @Override
            public void onSuccess(SendResult<String, Object> sendResult) {
                // 处理消息发送成功逻辑
                success.accept(sendResult);
            }
        });
    }

    public void send(String topic ,String key,Object payload) {
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic,key, payload);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable e) {
                log.error("--------消息发送失败------------");
            }

            @Override
            public void onSuccess(SendResult<String, Object> sendResult) {
                // 处理消息发送成功逻辑
                log.error("--------消息发送成功------------");
            }
        });
    }


    public void send(String topic ,Object payload) {
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(topic, payload);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable e) {
                log.error("--------消息发送失败------------");
            }

            @Override
            public void onSuccess(SendResult<String, Object> sendResult) {
                log.error("--------消息发送成功------------");
            }
        });
    }

    public void send(Object obj) {
        String obj2String = JSONObject.toJSONString(obj);
        log.info("准备发送消息为：{}", obj2String);
        //发送消息
        ListenableFuture<SendResult<String, Object>> future = kafkaTemplate.send(TOPIC_DEFAULT, obj);
        future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {
            @Override
            public void onFailure(Throwable e) {
                log.error("--------消息发送失败------------");
            }

            @Override
            public void onSuccess(SendResult<String, Object> sendResult) {
                log.error("--------消息发送成功------------");
            }
        });
    }
}
